package com.twd.reactor;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.nio.charset.StandardCharsets;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class ReactorServer {
    private static final Logger logger = LoggerFactory.getLogger(ReactorServer.class.getSimpleName());
    private Selector mainSelector;      // 主Selector
    private Selector[] subSelectors;    // 从Selector
    private int next = 0;
    private int count = 2;
    private boolean loop = true;
    private ServerSocketChannel serverChannel;

    public ReactorServer() {
        try {
            subSelectors = new Selector[count];         // 初始化多个从Selector，复用
            for (int i = 0; i < count; i++) {
                subSelectors[i] = Selector.open();
            }

            mainSelector = Selector.open();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public void init() {
        try {
            serverChannel = ServerSocketChannel.open();
            serverChannel.configureBlocking(false);
            ServerSocket socket = serverChannel.socket();
            socket.bind(new InetSocketAddress("localhost", 9999));
            SelectionKey selectionKey = serverChannel.register(mainSelector, SelectionKey.OP_ACCEPT);
            selectionKey.attach((Runnable) () -> {
                try {
                    System.out.println("建立连接");
                    SocketChannel client = serverChannel.accept();
                    new Handler(subSelectors[next], client);        // 每建立一个连接，就选择一个从Selector绑定
                    next++;
                    if (next == count) {
                        next = 0;
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });

            // 每个从Selector使用独立的线程监听事件，避免相互阻塞
            for (int i = 0; i < count; i++) {
                int finalI = i;
                new Thread(() -> {
                    new HandlerLoop(subSelectors[finalI]).run();
                }).start();
            }

            while (loop) {
                int select = mainSelector.select();
                if (select != 0) {
                    Set<SelectionKey> readKeys = mainSelector.selectedKeys();
                    Iterator<SelectionKey> iterator = readKeys.iterator();
                    while (iterator.hasNext()) {
                        SelectionKey key = iterator.next();
                        iterator.remove();
                        Runnable runnable = (Runnable) key.attachment();
                        System.out.println("-------");
                        runnable.run();
                    }
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    static class HandlerLoop implements Runnable {
        private Selector selector;

        public HandlerLoop(Selector selector) {
            this.selector = selector;
        }

        @Override
        public void run() {
            while (!Thread.interrupted()) {
                try {
                    logger.info("启动selector");
                    int select = selector.select();
                    System.out.println(select);
                    if (select != 0) {
                        Set<SelectionKey> readKeys = selector.selectedKeys();
                        Iterator<SelectionKey> iterator = readKeys.iterator();
                        while (iterator.hasNext()) {

                            SelectionKey key = iterator.next();
                            System.out.println("wwwwwwwwwwwwwwwwwww-------------------"+key);
                            Runnable runnable = (Runnable) key.attachment();
                            runnable.run();
                            iterator.remove();
                        }
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                }

                System.out.println("lllllllll");
            }
        }
    }


    static class Handler implements Runnable {
        private Selector selector;
        private SocketChannel socket;
        private SelectionKey selectionKey;
        private HandleState state;
        private ExecutorService pool;

        public Handler(Selector selector, SocketChannel socket) {

            this.selector = selector;
            this.socket = socket;
            this.state = HandleState.READING;
            this.pool = Executors.newFixedThreadPool(4);        // 增加一个线程池
            System.out.println(this.selector);
            System.out.println(this.socket);
            try {
                this.socket.configureBlocking(false);

                new Thread(()->{
                    try {
                        this.selectionKey = this.socket.register(this.selector, 0);
                        this.selectionKey.interestOps(SelectionKey.OP_READ);
                        this.selectionKey.attach(this);
                        logger.info("注册成功");
                    } catch (ClosedChannelException e) {
                        e.printStackTrace();
                    }
                }).start();
                TimeUnit.SECONDS.sleep(2);
                this.selector.wakeup();
            } catch (IOException | InterruptedException e) {
                e.printStackTrace();
            }
        }

        @Override
        public void run() {
            System.out.println("状态"+state);
            switch (state) {
                case READING:
                    state = HandleState.WORKING;        // 改变当前handler状态
                    pool.execute(() -> {                // 将耗时操作放在线程池中执行
//                        read();
                    });
                    break;
                case WRITING:
                    write();
                    break;
                default:
                    break;
            }
        }

        private void read() {
            StringBuffer sb = new StringBuffer();
            ByteBuffer buf = ByteBuffer.allocate(1024);
            try {
                while (true) {
                    buf.clear();
                    int read = socket.read(buf);
                    System.out.println("read");
                    sb.append(StandardCharsets.UTF_8.newDecoder().decode(buf.asReadOnlyBuffer()).toString());
                    if (read != 0) {
                        logger.info("receive message: {}.....", new String(buf.array(),0,buf.array().length));
                        Thread.sleep(2000);
//                        selectionKey.interestOps(SelectionKey.OP_READ);
//                        selectionKey.interestOps(SelectionKey.OP_WRITE);
//                        state = HandleState.WRITING;
                        state = HandleState.READING;
                        selector.wakeup();              // 唤醒Selector，让当前阻塞的selector.select()方法返回
                        break;
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        private void write() {
            try {
                ByteBuffer output = ByteBuffer.wrap("Reactor server answer!".getBytes());
                socket.write(output.duplicate());
            } catch (IOException e) {
                e.printStackTrace();
            } finally {
                selectionKey.cancel();
            }
        }

        private enum HandleState {
            WORKING,
            READING,
            WRITING
        }
    }
}